package com.group2.edu.realtime.dws.app;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.group2.edu.realtime.common.base.BaseApp;
import com.group2.edu.realtime.common.bean.Traffic4DimPageViewBean;
import com.group2.edu.realtime.common.function.BeanToJsonStrMapFunction;
import com.group2.edu.realtime.common.function.DimAsyncFunction;
import com.group2.edu.realtime.common.util.DateFormatUtil;
import com.group2.edu.realtime.common.util.FlinkSinkUtil;
import com.group2.edu.realtime.common.util.FlinkSourceUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.util.concurrent.TimeUnit;

/**
 * @author 高耀
 * @date 2024/12/16 16:50<p></p>
 * <p>
 * DWS层版本、来源、地区、是否是新访客四个维度页面浏览窗口信息统计
 * 需要启动的：
 * redis、maxwell、hdfs、Hbase、doris、flume1、zk、kafka、
 * DimApp、DwdBaseLog、DwsTrafficVcScArIsNewPageViewWindow、
 * DwdTrafficUniqueVisitorDetail
 */
public class DwsTrafficVcScArIsNewPageViewWindow extends BaseApp {
    public static void main(String[] args) {
        new DwsTrafficVcScArIsNewPageViewWindow().start(
                10022,
                4,
                "DwsTrafficVcScArIsNewPageViewWindow",
                "dwd_traffic_page"
        );
    }

    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> mainSource) {
        /**
         * 总思路：
         *      要统计四个维度下独立访客数、会话总数、页面浏览数、页面总停留时长，
         *      从dwd_traffic_page、dwd_traffic_unique_visitor_detail
         *      统一封装一个类用来装数据类型并统计，分别读取这2条数据流，转换成封装类型后union在一起，
         *      给水位线，
         *      然后分组开窗聚合，
         *      然后异步操作补充维度信息（省份信息和来源类型信息），先从redis取，取不到再到Hbase中取，再给redis缓存
         *      最后写入doris。
         */
        //TODO 1 读取dwd_traffic_unique_visitor_detail主题的数据
        DataStreamSource<String> uniqueVisitorDetailDs = env.fromSource(
                FlinkSourceUtil.getKafkaSource(
                        "dwd_traffic_unique_visitor_detail",
                        "DwsTrafficVcScArIsNewPageViewWindow"),
                WatermarkStrategy.noWatermarks(),
                "uniqueVisitorDetailDs"
        );

        //TODO 2 将2条流转换类型
        //2.1 转换page流，统计除独立访客外的统计字段
        SingleOutputStreamOperator<Traffic4DimPageViewBean> pageEncapsulatedDs = mainSource.map(
                new MapFunction<String, Traffic4DimPageViewBean>() {
                    @Override
                    public Traffic4DimPageViewBean map(String value) throws Exception {
                        //先将JSONStr串转成JSONObj
                        JSONObject jsonObj = JSON.parseObject(value);
                        //获取page对象
                        JSONObject pageObj = jsonObj.getJSONObject("page");
                        //获取common对象
                        JSONObject commonObj = jsonObj.getJSONObject("common");
                        //版本号
                        String vc = commonObj.getString("vc");
                        //来源id
                        String sc = commonObj.getString("sc");
                        //地区
                        String ar = commonObj.getString("ar");
                        //是否新访客
                        String isNew = commonObj.getString("is_new");
                        //时间戳
                        long ts = jsonObj.getLong("ts");
                        //使用建造者创建新对象返回
                        return Traffic4DimPageViewBean.builder()
                                .versionCode(vc)
                                .provinceId(ar)
                                .isNew(isNew)
                                .sourceId(sc)
                                .ts(ts)
                                .uvCount(0L)//独立访客都算0，因为要和独立访客流union，再独立访客流中统计
                                .totalSessionCount(
                                        StringUtils.isEmpty(pageObj.getString("last_page_id")) ?
                                                1L : 0L
                                )//如果没有上一页，就算一个独立会话，有就不算
                                .pageViewCount(1L)//肯定算1页
                                .totalDuringTime(pageObj.getLong("during_time")) //页面停留时长
                                .build();
                    }
                }
        );
        //2.2 转换独立访客流，统计独立访客字段
        SingleOutputStreamOperator<Traffic4DimPageViewBean> uvEncapsulatedDs = uniqueVisitorDetailDs.map(
                new MapFunction<String, Traffic4DimPageViewBean>() {
                    @Override
                    public Traffic4DimPageViewBean map(String value) throws Exception {
                        //先将JSONStr串转成JSONObj
                        JSONObject jsonObj = JSON.parseObject(value);
                        //获取page对象
                        JSONObject pageObj = jsonObj.getJSONObject("page");
                        //获取common对象
                        JSONObject commonObj = jsonObj.getJSONObject("common");
                        //版本号
                        String vc = commonObj.getString("vc");
                        //来源id
                        String sc = commonObj.getString("sc");
                        //地区
                        String ar = commonObj.getString("ar");
                        //是否新访客
                        String isNew = commonObj.getString("is_new");
                        //时间戳
                        long ts = jsonObj.getLong("ts");
                        //使用建造者创建新对象返回
                        return Traffic4DimPageViewBean.builder()
                                .versionCode(vc)
                                .provinceId(ar)
                                .isNew(isNew)
                                .sourceId(sc)
                                .ts(ts)
                                .uvCount(1L)//独立访客都算1，其他都算0
                                .totalSessionCount(0L)
                                .pageViewCount(0L)
                                .totalDuringTime(0L)
                                .build();
                    }
                }
        );


        //TODO 3 合流
        DataStream<Traffic4DimPageViewBean> unionDs
                = pageEncapsulatedDs.union(uvEncapsulatedDs);

        //TODO 4 指定水位线、以指定维度分组、开滚动窗口、聚合操作
        //4.1 指定水位线生成策略
        SingleOutputStreamOperator<Traffic4DimPageViewBean> withWmDs = unionDs.assignTimestampsAndWatermarks(
                WatermarkStrategy
                        .<Traffic4DimPageViewBean>forMonotonousTimestamps()
                        .withTimestampAssigner((bean, ts) -> bean.getTs())

        );
        // 合流以后没问题
//        withWmDs.print("合流");

        //4.2 分组
        KeyedStream<Traffic4DimPageViewBean, String> keyBy4DimDs = withWmDs.keyBy(
                new KeySelector<Traffic4DimPageViewBean, String>() {
                    @Override
                    public String getKey(Traffic4DimPageViewBean bean) throws Exception {
                        //以四个维度组成的四元组为分组依据
                        return bean.getSourceId() +
                                bean.getProvinceId() +
                                bean.getIsNew() +
                                bean.getVersionCode();
                    }
                }
        );
        //4.3 开窗
        WindowedStream<Traffic4DimPageViewBean, String, TimeWindow> windowedDs
                = keyBy4DimDs.window(TumblingEventTimeWindows.of(Time.seconds(10)));
        //4.4 聚合，因为进出类型相同，使用reduce即可，不用process
        SingleOutputStreamOperator<Traffic4DimPageViewBean> reducedDs = windowedDs.reduce(
                //聚合函数，进行聚合操作
                new ReduceFunction<Traffic4DimPageViewBean>() {
                    @Override
                    public Traffic4DimPageViewBean reduce(
                            Traffic4DimPageViewBean v1,
                            Traffic4DimPageViewBean v2
                    ) throws Exception {
                        //相加
                        v1.setPageViewCount(v1.getPageViewCount() + v2.getPageViewCount());
                        v1.setTotalDuringTime(v1.getUvCount() + v2.getUvCount());
                        v1.setTotalSessionCount(v1.getTotalSessionCount() + v2.getTotalSessionCount());
                        v1.setTotalDuringTime(v1.getTotalDuringTime() + v2.getTotalDuringTime());
                        //返回reduce结果
                        return v1;
                    }
                },
                //窗口函数，补全窗口信息
                new ProcessWindowFunction<Traffic4DimPageViewBean, Traffic4DimPageViewBean, String, TimeWindow>() {

                    @Override
                    public void process(
                            String s,
                            ProcessWindowFunction<Traffic4DimPageViewBean, Traffic4DimPageViewBean, String, TimeWindow>.Context context,
                            Iterable<Traffic4DimPageViewBean> iterable,
                            Collector<Traffic4DimPageViewBean> collector
                    ) throws Exception {
                        //获取窗口对象
                        TimeWindow window = context.window();
                        //获取起止日期
                        String startDateTime = DateFormatUtil.tsToDateTime(window.getStart());
                        String endDateTime = DateFormatUtil.tsToDateTime(window.getEnd());
                        String curDate = DateFormatUtil.tsToDate(window.getStart());
                        Traffic4DimPageViewBean bean = iterable.iterator().next();

                        bean.setStt(startDateTime);
                        bean.setEdt(endDateTime);
                        //设置cur_date
                        bean.setCurDate(curDate);
                        //修正时间戳
                        bean.setTs(System.currentTimeMillis());
                        //写出
                        collector.collect(bean);


                    }
                }
        ).setParallelism(2);
        //测试
//        reducedDs.print();

        //TODO 5 补充维度信息
        SingleOutputStreamOperator<Traffic4DimPageViewBean> withProvinceDimDs = AsyncDataStream.unorderedWait(
                reducedDs,
                new DimAsyncFunction<Traffic4DimPageViewBean>() {

                    @Override
                    public void addDims(Traffic4DimPageViewBean obj, JSONObject dimJsonObj) {
                        //赋值字段
                        obj.setProvinceName(dimJsonObj.getString("name"));
                    }

                    @Override
                    public String getTableName() {
                        //Hbase中表名
                        return "dim_base_province";
                    }

                    @Override
                    public String getRowKey(Traffic4DimPageViewBean obj) {
                        //查询的根据id
                        return obj.getProvinceId();
                    }
                },
                120,
                TimeUnit.SECONDS
        );

        //TODO 6 补充来源维度信息
        SingleOutputStreamOperator<Traffic4DimPageViewBean> withSourceDimDs = AsyncDataStream.unorderedWait(
                withProvinceDimDs,
                new DimAsyncFunction<Traffic4DimPageViewBean>() {

                    @Override
                    public void addDims(Traffic4DimPageViewBean obj, JSONObject dimJsonObj) {
                        //赋值字段
                        obj.setSourceName(dimJsonObj.getString("source_site"));
                    }

                    @Override
                    public String getTableName() {
                        //Hbase中表名
                        return "dim_base_source";
                    }

                    @Override
                    public String getRowKey(Traffic4DimPageViewBean obj) {
                        //查询的根据id
                        return obj.getSourceId();
                    }
                },
                120,
                TimeUnit.SECONDS
        );
        //测试
        withSourceDimDs.print("聚合后");

        //TODO 7 写入doris
        SingleOutputStreamOperator<String> mappedDs = withSourceDimDs
                .map(new BeanToJsonStrMapFunction<>());//先转成JSON串
        mappedDs.print();
        mappedDs.sinkTo(FlinkSinkUtil.getDorisSink(
                "dws_traffic_vc_source_ar_is_new_page_view_window"));//写入
    }
}
